package ThreadPool_26;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义的阻塞队列
 */
public class MyBlockingQueue<T> {

    // 使用list作为队列存储
    private final List<T> queue;

    // 阻塞队列的最大长度
    private final int limit;

    // 队列名
    private final String queueName;

    // 入列次数
    private AtomicInteger enqueueTimes = new AtomicInteger();

    // 出列次数
    private AtomicInteger dequeueTimes = new AtomicInteger();

    // 队列数量
    private AtomicInteger queueCount = new AtomicInteger();

    // 是否停止
    private boolean isStop;

    private Lock lock = new ReentrantLock();

    // 初始化阻塞队列
    public MyBlockingQueue(int limit, String queueName) {
        this.queue = new CopyOnWriteArrayList<>();
        this.limit = limit;
        this.queueName = queueName;
    }

    public AtomicInteger getEnqueueTimes() {
        return enqueueTimes;
    }

    /**
     * 入队
     * @param item
     * @throws InterruptedException
     */
    public void enqueue(T item) throws InterruptedException {

        // 判断，如果当前队列已满，则入列阻塞
        lock.lock();
        if (this.queueCount.get() == this.limit) {
            System.out.println("【 "+ queueName+ "】入列等待：" + Thread.currentThread());
            lock.wait();
        }

        lock.notifyAll();
        lock.unlock();
        System.out.println("【 "+ queueName+ "】入列" + this.enqueueTimes.addAndGet(1) +"\t\t" + "当前线程：" + Thread.currentThread());
        this.queue.add(item);
        this.queueCount.getAndIncrement();
    }

    /**
     * 出队
     */
    public T dequeue() throws InterruptedException {

        // 判断，如果当前队列为空，则出列阻塞
        lock.lock();
        while (queueCount.get() == 0  && !isStop) {
            System.out.println("【 "+ queueName+ "】出列等待：" + Thread.currentThread());
            lock.wait();
        }

        lock.notifyAll();
        lock.unlock();
        if(isStop) {
            System.out.println("【 "+ queueName+ "】出列停止等待：" + Thread.currentThread());
            return null;
        }

        // 弹出第一个
        T remove = this.queue.remove(0);
        int i = this.dequeueTimes.addAndGet(1);
        this.queueCount.getAndDecrement();
        System.out.println("【 "+ queueName+ "】出列"  +  i +"\t\t" + "当前线程：" + Thread.currentThread());

        return remove;
    }

    public synchronized void doStop() {
        this.isStop = true;

        // 唤醒所有等待出列线程
        this.notifyAll();
    }

    public static void main(String[] args) {

        MyBlockingQueue<Integer> myBlockingQueue = new MyBlockingQueue<>(30, "数据队列");

        for (int i = 1; i <= 50; i++) {
            Integer finalI = i;
            new Thread(() -> {
                try {
                    System.out.println("出列：" + myBlockingQueue.dequeue());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        // 入队
        for(int i = 1; i <= 50; i++) {
            Integer finalI = i;
            new Thread(() -> {
                try {
                    System.out.println("入列：" + finalI);
                    myBlockingQueue.enqueue(finalI);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        System.out.println("业务----------");
    }
}